package com.fanxuankai.boot.mqbroker.consumer;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.core.net.NetUtil;
import cn.hutool.core.thread.ThreadUtil;
import com.alibaba.fastjson.JSON;
import com.fanxuankai.boot.mqbroker.Event;
import com.fanxuankai.boot.mqbroker.MqBrokerDingTalkClientHelper;
import com.fanxuankai.boot.mqbroker.autoconfigure.MqBrokerProperties;
import com.fanxuankai.boot.mqbroker.consumer.autoconfigure.MqBrokerConsumerProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * @author fanxuankai
 */
public class MsgReceiveConsumer {
    private static final Logger LOGGER = LoggerFactory.getLogger(MsgReceiveConsumer.class);
    @Resource
    private MsgReceiveService msgReceiveService;
    @Resource
    private EventDistributorFactory eventDistributorFactory;
    @Resource
    private MqBrokerProperties brokerProperties;
    @Resource
    private MqBrokerConsumerProperties properties;
    @Resource
    private DataSourceTransactionManager dataSourceTransactionManager;
    @Resource
    private TransactionDefinition transactionDefinition;
    @Resource
    private EventListenerRegistry eventListenerRegistry;
    private final MqBrokerDingTalkClientHelper mqBrokerDingTalkClientHelper;

    public MsgReceiveConsumer(MqBrokerDingTalkClientHelper mqBrokerDingTalkClientHelper) {
        this.mqBrokerDingTalkClientHelper = mqBrokerDingTalkClientHelper;
    }

    /**
     * 消费消息
     *
     * @param msg   消息
     * @param retry 是否重试
     */
    public void consume(MsgReceive msg, boolean retry) {
        int i = msg.getRetry();
        AbstractEventDistributor abstractEventDistributor = eventDistributorFactory.get(msg);
        while (true) {
            TransactionStatus transactionStatus = null;
            try {
                msg = BeanUtil.copyProperties(msg, MsgReceive.class);
                Event<Object> event = new Event<>();
                event.setName(msg.getTopic());
                event.setKey(msg.getCode());
                Class<?> dataType = eventListenerRegistry.getDataType(msg.getTopic());
                if (dataType == String.class) {
                    event.setData(msg.getData());
                } else {
                    event.setData(JSON.parseObject(msg.getData(), dataType));
                }
                // 手动开启事务
                transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
                long start = System.currentTimeMillis();
                abstractEventDistributor.distribute(event);
                long time = System.currentTimeMillis() - start;
                msgReceiveService.success(msg);
                // 手动提交事务
                dataSourceTransactionManager.commit(transactionStatus);
                if (properties.getSlowThreshold() != null && time > properties.getSlowThreshold()) {
                    mqBrokerDingTalkClientHelper.push("消息消费过慢", msg.getTopic(), msg.getCode(),
                            time, NetUtil.getLocalhostStr());
                }
                LOGGER.info("消息消费成功, topic: {}, code: {}, 耗时: {}ms", msg.getTopic(), msg.getCode(), time);
                break;
            } catch (Throwable throwable) {
                // 手动回滚事务
                if (transactionStatus != null) {
                    dataSourceTransactionManager.rollback(transactionStatus);
                }
                msg.setCause(ExceptionUtil.stacktraceToString(throwable));
                msg.setRetry(++i);
                if (retry && i < properties.getMaxRetry()) {
                    ThreadUtil.sleep(properties.getRetryInterval(), TimeUnit.MILLISECONDS);
                } else {
                    msgReceiveService.failure(msg);
                    LOGGER.error("消息消费失败, topic: {}, code: {}", msg.getTopic(), msg.getCode(), throwable);
                    if (mqBrokerDingTalkClientHelper != null) {
                        mqBrokerDingTalkClientHelper.push("消息消费失败", msg.getTopic(), msg.getCode(),
                                msg.getRetry(), NetUtil.getLocalhostStr());
                    }
                    break;
                }
            }
        }
    }
}
